package factories

import (
	"encoding/json"
	"errors"
	"net"
	"sync"

	"github.com/IBM/sarama"
	"hyper/supports/configs"
	"hyper/supports/logger"
	"hyper/supports/queue/abstruct"
)

type KafkaQueue struct {
	Producer  sarama.SyncProducer
	Data      *abstruct.QueueObj
	JsonData  string
	partition int32
	offset    int64
}

// 消费者回调方法
type KafkaConsumer func(msg *sarama.ConsumerMessage)

func (k *KafkaQueue) New(obj *abstruct.QueueObj) *KafkaQueue {
	// 配置 Kafka 生产者
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有副本确认
	config.Producer.Retry.Max = 5                    // 重试次数
	config.Producer.Return.Successes = true          // 返回成功消息

	// 创建 Kafka 生产者
	producer, err := sarama.NewSyncProducer([]string{kafkaHost()}, config)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "创建生产者时出错",
		})
	}

	// 连接负值方便后期关闭
	k.Producer = producer

	// 队列参数赋值
	k.Data = obj

	return k
}

// 将结构体转为Json
func (k *KafkaQueue) MakeJson() *KafkaQueue {
	// 获取格式化后得 json 字符串
	jsonStr, err := json.Marshal(k.Data)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "转换json 字符串失败",
		})
	}

	k.JsonData = string(jsonStr)

	// 获取队列名称
	k.Data.Name = abstruct.QueueName(k.Data.Name)

	return k

}

// 压入队列 并返回结果
func (k *KafkaQueue) Producers() error {

	// 发送消息
	msg := &sarama.ProducerMessage{
		Topic: k.Data.Name,
		Key:   sarama.StringEncoder(k.Data.Name),
		Value: sarama.StringEncoder(k.JsonData),
	}
	// 关闭链接与频道链接
	k.Close()

	partition, offset, err := k.Producer.SendMessage(msg)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "压入队列失败",
		})
		return errors.New("压入队列失败: " + err.Error())
	}

	k.partition = partition
	k.offset = offset

	return nil
}

// 关闭链接
func (k *KafkaQueue) Close() {
	k.Producer.Close()
}

// 消费者代码
func (k *KafkaQueue) Consumer(fn KafkaConsumer) {
	// 配置 Kafka 消费者
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.AutoCommit.Enable = true      // 禁用自动提交偏移量
	config.Consumer.Offsets.Initial = sarama.OffsetOldest // 从最早的偏移量开始消费

	// 创建 Kafka 消费者
	consumer, err := sarama.NewConsumer([]string{kafkaHost()}, config)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "创建消费者时出错",
		})
	}
	defer consumer.Close()

	// 订阅主题
	partition, err := consumer.ConsumePartition(k.Data.Name, k.partition, sarama.OffsetNewest)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "创建分区使用者时出错",
		})
	}
	defer partition.Close()

	var wg sync.WaitGroup
	for message := range partition.Messages() {
		wg.Add(1)

		// 处理消息
		fn(message)
	}

	wg.Wait()
}

// kafka主机地址
func kafkaHost() string {
	return net.JoinHostPort(configs.GetString("queue.kafka.host"), configs.GetString("queue.kafka.port"))
}
